/*
 *
 * Copyright (c) 2020 The Raptor Authors. All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

#include "src/windows/tcp_connector.h"

#include <string.h>

#include "raptor-lite/impl/connector.h"
#include "raptor-lite/impl/config.h"
#include "raptor-lite/impl/endpoint.h"
#include "raptor-lite/utils/log.h"
#include "src/common/endpoint_impl.h"
#include "src/common/socket_util.h"
#include "src/windows/socket_settings.h"

namespace raptor {
struct AsyncConnectObject {
    ListEntry entry;
    OverLappedEx ole;
    raptor_resolved_address addr;
    SOCKET fd;
    intptr_t user_value;
};

TcpConnector::TcpConnector(ConnectorHandler *handler)
    : service_(handler)
    , shutdown_(true)
    , connectex_(nullptr)
    , tcp_user_timeout_ms_(0)
    , poll_thread_(nullptr) {
    RAPTOR_LIST_INIT(&list_head_);
}

TcpConnector::~TcpConnector() { Shutdown(); }

raptor_error TcpConnector::Init(PollingThread *poll_thread, int tcp_user_timeout) {
    if (!shutdown_) return RAPTOR_ERROR_FROM_STATIC_STRING("TcpConnector has been initialized");

    bool success = false;
    thd_ = Thread("TcpListener", std::bind(&TcpConnector::WorkThread, this, std::placeholders::_1),
                  nullptr, &success);
    if (!success) {
        return RAPTOR_ERROR_FROM_STATIC_STRING("TcpConnector: Failed to create thread");
    }

    shutdown_ = false;
    poll_thread_ = poll_thread;
    tcp_user_timeout_ms_ = tcp_user_timeout;
    count_.Store(0);
    return RAPTOR_ERROR_NONE;
}

raptor_error TcpConnector::Start() {
    if (shutdown_) {
        return RAPTOR_ERROR_FROM_STATIC_STRING("TcpConnector has not been initialized");
    }
    thd_.Start();
    return RAPTOR_ERROR_NONE;
}

void TcpConnector::Shutdown() {
    if (!shutdown_) {
        log_warn("TcpConnector: prepare to shutdown");
        shutdown_ = true;
        cv_.Signal();
        thd_.Join();

        mtx_.Lock();
        ListEntry *entry = nullptr;
        do {
            entry = ListPopBack(&list_head_);
            if (entry) {
                auto obj = reinterpret_cast<AsyncConnectObject *>(entry);
                poll_thread_->Delete(obj->fd);
                closesocket(obj->fd);
                delete obj;
            }
        } while (entry);
        mtx_.Unlock();

        // clear message queue
        bool empty = true;
        do {
            auto n = mpscq_.PopAndCheckEnd(&empty);
            auto msg = reinterpret_cast<PollingThreadMessageNode *>(n);
            if (msg != nullptr) {
                delete msg;
            }
        } while (!empty);

        log_warn("TcpConnector: shutdown");
    }
}

raptor_error TcpConnector::Connect(const std::string &addr, intptr_t user) {
    raptor_resolved_addresses *addrs;
    auto e = raptor_blocking_resolve_address(addr.c_str(), nullptr, &addrs);
    if (e != RAPTOR_ERROR_NONE) {
        return e;
    }
    if (addrs->naddrs == 0) {
        return RAPTOR_ERROR_FROM_FORMAT("Invalid address: (%s)", addr.c_str());
    }

    e = InternalConnect(&addrs->addrs[0], user, tcp_user_timeout_ms_);
    if (e != RAPTOR_ERROR_NONE) {
        log_warn("TcpConnector: Failed to connect %s, %s", addr.c_str(), e->ToString().c_str());
    }
    raptor_resolved_addresses_destroy(addrs);
    return e;
}

AsyncConnectObject *TcpConnector::FindRecordEntryFromOverlapped(OverLappedEx *overlaped) {
    intptr_t olex = reinterpret_cast<intptr_t>(overlaped);
    size_t offset = offsetof(AsyncConnectObject, ole);
    intptr_t entry = olex - static_cast<intptr_t>(offset);
    return reinterpret_cast<AsyncConnectObject *>(entry);
}

void TcpConnector::OnEventProcessImpl(EventDetail *detail) {

    AsyncConnectObject *entry = FindRecordEntryFromOverlapped(detail->overlaped);

    // get local address
    raptor_resolved_address local;
    local.len = sizeof(local.addr);
    memset(local.addr, 0, local.len);
    getsockname(entry->fd, (struct sockaddr *)local.addr, (int *)&local.len);

    // remove monitoring of fd from IOCP
    poll_thread_->Delete(entry->fd);

    std::shared_ptr<EndpointImpl> endpoint =
        std::make_shared<EndpointImpl>(entry->fd, &local, &entry->addr);

    if (detail->event_type & internal::kErrorEvent) {
        // Maybe an error occurred or the connection was closed
        raptor_error err = RAPTOR_WINDOWS_ERROR(detail->error_code, "IOCP_WAIT");
        service_->OnConnectFailed(entry->user_value, err);
    } else {
        // update connect context
        setsockopt(entry->fd, SOL_SOCKET, SO_UPDATE_CONNECT_CONTEXT, NULL, 0);
        Property property({config::kUserCustomValue, entry->user_value});
        service_->OnConnect(endpoint, property);
        if (endpoint->IsValid() && property.Count() > 1) {
            ProcessProperty(entry->fd, property);
        }
    }

    AutoMutex g(&mtx_);
    ListRemoveEntry(&entry->entry);
    delete entry;
}

void TcpConnector::ProcessProperty(SOCKET fd, const Property &p) {
    bool SocketNoSIGPIPE = true;
    if (p.CheckValue<bool>(config::kSocketNoSIGPIPE, SocketNoSIGPIPE) && SocketNoSIGPIPE) {
        raptor_set_socket_no_sigpipe_if_possible(fd);
    }

    bool SocketReuseAddress = true;
    if (p.CheckValue<bool>(config::kSocketReuseAddress, SocketReuseAddress) &&
        !SocketReuseAddress) {
        raptor_set_socket_reuse_addr(fd, 0);
    }

    bool SocketReusePort = false;
    if (p.CheckValue<bool>(config::kSocketReusePort, SocketReusePort) && SocketReusePort) {
        raptor_set_socket_reuse_port(fd, 1);
    }

    bool SocketLowLatency = true;
    if (p.CheckValue<bool>(config::kSocketLowLatency, SocketLowLatency) && !SocketLowLatency) {
        raptor_set_socket_low_latency(fd, 0);
    }

    int SocketSendTimeoutMs = 0;
    if (p.CheckValue<int>(config::kSocketSendTimeoutMs, SocketSendTimeoutMs) &&
        SocketSendTimeoutMs > 0) {
        raptor_set_socket_snd_timeout(fd, SocketSendTimeoutMs);
    }

    int SocketRecvTimeoutMs = 0;
    if (p.CheckValue<int>(config::kSocketRecvTimeoutMs, SocketRecvTimeoutMs) &&
        SocketRecvTimeoutMs > 0) {
        raptor_set_socket_rcv_timeout(fd, SocketRecvTimeoutMs);
    }

    bool SocketNonBlocking = true;
    if (p.CheckValue<bool>(config::kSocketNonBlocking, SocketNonBlocking) && !SocketNonBlocking) {
        raptor_set_socket_nonblocking(fd, 0);
    }
}

raptor_error TcpConnector::GetConnectExIfNecessary(SOCKET s) {
    if (!connectex_) {
        GUID guid = WSAID_CONNECTEX;
        DWORD ioctl_num_bytes;
        int status;

        /* Grab the function pointer for ConnectEx for that specific socket.
            It may change depending on the interface. */
        status = WSAIoctl(s, SIO_GET_EXTENSION_FUNCTION_POINTER, &guid, sizeof(guid), &connectex_,
                          sizeof(connectex_), &ioctl_num_bytes, NULL, NULL);

        if (status != 0) {
            return RAPTOR_WINDOWS_ERROR(WSAGetLastError(),
                                        "WSAIoctl(SIO_GET_EXTENSION_FUNCTION_POINTER)");
        }
    }

    return RAPTOR_ERROR_NONE;
}

raptor_error TcpConnector::InternalConnect(const raptor_resolved_address *addr, intptr_t user,
                                           int timeout_millseconds) {
    raptor_dualstack_mode mode = RAPTOR_DSMODE_NONE;
    raptor_resolved_address local_address;
    raptor_resolved_address mapped_addr;
    int status;
    BOOL ret;
    char *str_addr = nullptr;

    struct AsyncConnectObject *entry = new struct AsyncConnectObject;

    // SHADOW: If OVERLAPPED not initialized, ConnectEx will fail
    // and WSAGetLastError will return 6.
    memset(entry, 0, sizeof(*entry));

    raptor_error error = raptor_create_socket(addr, &mapped_addr, &entry->fd, &mode);

    if (error != RAPTOR_ERROR_NONE) {
        goto failure;
    }
    error = raptor_tcp_prepare_socket(entry->fd, timeout_millseconds);
    if (error != RAPTOR_ERROR_NONE) {
        goto failure;
    }
    error = GetConnectExIfNecessary(entry->fd);
    if (error != RAPTOR_ERROR_NONE) {
        goto failure;
    }

    raptor_sockaddr_make_wildcard6(0, &local_address);

    status = bind(entry->fd, (raptor_sockaddr *)&local_address.addr, (int)local_address.len);

    if (status != 0) {
        error = RAPTOR_WINDOWS_ERROR(WSAGetLastError(), "bind");
        goto failure;
    }

    RAPTOR_LIST_ENTRY_INIT(&entry->entry);
    entry->ole.event_type = internal::kConnectEvent;
    entry->ole.user_id = 0;
    entry->user_value = user;

    mtx_.Lock();
    memcpy(&entry->addr, &mapped_addr, sizeof(mapped_addr));
    ListPushBack(&list_head_, &entry->entry);
    poll_thread_->Add(entry->fd, this);
    mtx_.Unlock();

    ret = connectex_(entry->fd, (raptor_sockaddr *)&mapped_addr.addr, (int)mapped_addr.len, NULL, 0,
                     NULL, &entry->ole.overlapped);

    /* It wouldn't be unusual to get a success immediately. But we'll still get
        an IOCP notification, so let's ignore it. */
    if (!ret) {
        int last_error = WSAGetLastError();
        if (last_error != ERROR_IO_PENDING) {
            error = RAPTOR_WINDOWS_ERROR(WSAGetLastError(), "ConnectEx");
            goto failure;
        }
    }

    raptor_sockaddr_to_string(&str_addr, &mapped_addr, 0);
    if (str_addr) {
        log_info("TcpConnector: start connecting %s", str_addr);
        free(str_addr);
    }
    return RAPTOR_ERROR_NONE;

failure:
    if (entry->fd != INVALID_SOCKET) {
        closesocket(entry->fd);
    }
    delete entry;
    return error;
}

void TcpConnector::OnEventProcess(EventDetail *detail) {
    auto msg = new PollingThreadMessageNode;
    memcpy(&msg->detail, detail, sizeof(*detail));
    mpscq_.push(&msg->node);
    count_.FetchAdd(1, MemoryOrder::RELAXED);
    cv_.Signal();
}

void TcpConnector::WorkThread(void *) {
    while (!shutdown_) {
        if (count_.Load() < 1) {
            AutoMutex g(&cmtx_);
            cv_.Wait(&cmtx_, 1000);
            continue;
        }
        auto n = mpscq_.pop();
        auto msg = reinterpret_cast<PollingThreadMessageNode *>(n);
        if (msg != nullptr) {
            count_.FetchSub(1, MemoryOrder::RELAXED);
            this->OnEventProcessImpl(&msg->detail);
            delete msg;
        }
    }
}

} // namespace raptor
